package com.redstar.HappyRefresh.service.impl;

import com.redstar.HappyRefresh.constant.RedisConstant;
import com.redstar.HappyRefresh.pojo.*;
import com.redstar.HappyRefresh.service.InterestPushService;
import com.redstar.HappyRefresh.service.TypeService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.util.ObjectUtils;

import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

@Service
@Slf4j
public class InterestPushServiceIMpl implements InterestPushService { // 兴趣推服务

    @Autowired
    private TypeService typeService;
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    @Override
    @Async
    public void initUserModel(Long userId, List<String> typeLabels) { // // 用户订阅分类后重新初始化模型
        // 因为是异步调用，所以userId需要通过参数传递
        HashMap<String, Double> userMap = new HashMap<>(); // key为标签名，value为概率，每个标签的概率都是相同的
        if (!ObjectUtils.isEmpty(typeLabels)) {
            double probabilityValue = 100.0 / typeLabels.size();
            for (String labelName : typeLabels) { // 存在相同的标签也不会有影响，因为每个标签的概率是相同的
                userMap.put(labelName, probabilityValue);
            }
        }
        // 用户模型是记录在redis中，因为用户模型需要频繁修改
        String key = RedisConstant.USER_MODEL + userId;
        // 将当前用户模型存入redis，覆盖之前的用户模型
        redisTemplate.opsForHash().putAll(key, userMap);
        redisTemplate.expire(key, RedisConstant.USER_MODEL_TIME, TimeUnit.SECONDS);
    }

    @Override
    @Async
    public void updateUserModel(UserModel userModel) { // 修改用户模型
        // 1.观看浏览量到达总时长1/5  +1概率
        // 2.观看浏览量未到总时长1/5 -0.5概率
        // 3.点赞视频  +2概率
        // 4.收藏视频  +3概率
        String key = RedisConstant.USER_MODEL + userModel.getUserId();
        Map<Object, Object> labelMap = redisTemplate.opsForHash().entries(key);
        if (ObjectUtils.isEmpty(labelMap)) {
            labelMap = new HashMap<>();
        }
        for (Model model : userModel.getModels()) {
            String label = model.getLabels();
            Double score = model.getScore();
            if (labelMap.containsKey(label)) {
                labelMap.put(label, Double.parseDouble(labelMap.get(label).toString()) + score);
                double newScore = Double.parseDouble(labelMap.get(label).toString());
                if (newScore <= 0) {
                    labelMap.remove(label);
                }
            } else if (score > 0) {
                labelMap.put(label, score);
            }
        }
        // 每个标签概率同等加上标签数，再同等除以标签数  防止数据膨胀
        int labelSize = labelMap.size();
        labelMap.replaceAll((k, v) -> (Double.parseDouble(v.toString()) + labelSize) / labelSize);
        // 更新用户模型
        redisTemplate.opsForHash().putAll(key, labelMap);
        redisTemplate.expire(key, RedisConstant.USER_MODEL_TIME, TimeUnit.SECONDS);
    }

    @Override
    public List<Long> listVideoIdByUserModel(User user) { // 根据用户模型推送10条视频
        Long userId = user.getId();
        List<Long> resultVideoIdList = new ArrayList<>();
        if (userId != null) {
            String key1 = RedisConstant.USER_MODEL + userId;
            Map<Object, Object> modelMap = redisTemplate.opsForHash().entries(key1); // 获取所有标签和推送概率
            if (!ObjectUtils.isEmpty(modelMap)) {
                // 将map转化为String数组
                String[] modelArray = initProbabilityArray(modelMap);
                // 从String数组中随机获取9个标签
                ArrayList<String> labelList = new ArrayList<>();
                Random random = new Random();
                for (int i = 0; i < 9; i++) {
                    labelList.add(modelArray[random.nextInt(modelArray.length)]);
                }
                // 从系统视频库中获取视频，所有公开的视频都在里面，key为system:stock:label，value为视频id
                String key2 = RedisConstant.SYSTEM_STOCK;
                List<Object> videoIdList = redisTemplate.executePipelined((RedisCallback<?>) connection -> {
                    for (String label : labelList) {
                        String key = key2 + label;
                        connection.sRandMember(key.getBytes());
                    }
                    return null;
                });
                // 可能会有重复，转化为set，起到去重的效果
                Set<Long> videoIdSet = videoIdList.stream().filter(o -> !ObjectUtils.isEmpty(o)).map(o -> Long.valueOf(o.toString())).collect(Collectors.toSet());
                // 为避免重复推送，获取用户已浏览过的视频的id，用于去重
                String key3 = RedisConstant.HISTORY_VIDEO;
                List<Object> repeatVideoIdList = redisTemplate.executePipelined((RedisCallback<?>) connection -> {
                    for (Long videoId : videoIdSet) {
                        // key为history:video:${videoId}:${userId}，value为videoId
                        String key = key3 + videoId + ":" + userId;
                        connection.get(key.getBytes());
                    }
                    return null;
                });
                Set<Long> repeatVideoIdSet = repeatVideoIdList.stream().filter(o -> !ObjectUtils.isEmpty(o)).map(o -> Long.valueOf(o.toString())).collect(Collectors.toSet());
                // 去除当前用户已浏览过的视频的id
                videoIdSet.removeAll(repeatVideoIdSet);
                // 将处理好的视频id加入结果集
                resultVideoIdList.addAll(videoIdSet);
                // 前面只取了9个标签，所以最多只有9个视频id，最后1个根据性别推送，这是为了避免返回的resultVideoIdList的内容为null
                Boolean sex = user.getSex();
                Long lastRandomVideoId = randomSexVideoId(sex);
                resultVideoIdList.add(lastRandomVideoId);
                return resultVideoIdList;
            }
        }
        // 游客登录或用户模型为null，随机推送10条视频
        // 随机获取10个标签，可能会有重复
        List<String> labelList = typeService.random10Labels();
        // 将标签与常量拼装起来，组成的是系统视频库的key
        ArrayList<String> videoIdKey = new ArrayList<>();
        String key = RedisConstant.SYSTEM_STOCK;
        for (String label : labelList) {
            videoIdKey.add(key + label);
        }
        List<Object> list = redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
            for (String k : videoIdKey) {
                connection.sRandMember(k.getBytes()); // 从set中随机获取一个元素
            }
            return null;
        });
        List<Object> objectList = new ArrayList<>();
        for (Object o : list) {
            if (o != null) objectList.add(o);
        }
        if (!ObjectUtils.isEmpty(objectList)) { // objectList为null，说明随机标签对应的key中没有数据
            // 去重，避免推送重复视频
            Set<Long> videoIdSet = objectList.stream().map(o -> Long.valueOf(o.toString())).collect(Collectors.toSet());
            resultVideoIdList.addAll(videoIdSet);
        }
        return resultVideoIdList;
    }

    private Long randomSexVideoId(Boolean sex) { // 根据用户的性别随机推送视频
        // 男生为体育明星，女生为明星八卦
        String key = RedisConstant.SYSTEM_STOCK + (sex ? "体育明星" : "明星八卦"); // key为system:stock:${label}，value为videoId
        Object videoId = redisTemplate.opsForSet().randomMember(key);
        return Long.valueOf(videoId.toString());
    }

    private String[] initProbabilityArray(Map<Object, Object> modelMap) { // 将map转化为数组
        HashMap<String, Integer> probabilityMap = new HashMap<>();
        AtomicInteger total = new AtomicInteger(0);
        int size = modelMap.size();
        modelMap.forEach((k, v) -> {
            // 将标签的概率转化为数组中的数量
            int num = (((Double) v).intValue() + size) / size;
            total.addAndGet(num);
            probabilityMap.put(k.toString(), num);
        });
        String[] modalArray = new String[total.get()];
        AtomicInteger index = new AtomicInteger(0);
        probabilityMap.forEach((k, v) -> {
            int i = index.get();
            for (int j = 0; j < v; j++) {
                modalArray[i + j] = k;
            }
            index.addAndGet(v);
        });
        return modalArray;
    }

    @Override
    public Set<Long> listVideoIdByTypeId(Long typeId) { // 根据视频分类推送10条视频
        // 视频存储在redis的系统分类库中
        List<Object> objectList = redisTemplate.opsForSet().randomMembers(RedisConstant.SYSTEM_TYPE_STOCK + typeId, 10);
        // 将list转化set时为了去重
        Set<Long> videoIdSet = objectList.stream().filter(o -> !ObjectUtils.isEmpty(o)).map(o -> Long.valueOf(o.toString())).collect(Collectors.toSet());
        if (ObjectUtils.isEmpty(videoIdSet)) {
            log.error("typeId对应的redis分类库中没有数据");
        }
        return videoIdSet;
    }

    @Override
    public Set<Long> listVideoIdByLabels(ArrayList<String> labelList) { // 根据视频标签推送视频
        List<Object> objectList = redisTemplate.executePipelined((RedisCallback<?>) connection -> {
            for (String label : labelList) {
                String key = RedisConstant.SYSTEM_STOCK + label;
                connection.sRandMember(key.getBytes());
            }
            return null;
        });
        Set<Long> videoIdSet = objectList.stream().filter(o -> !ObjectUtils.isEmpty(o)).map(o -> Long.valueOf(o.toString())).collect(Collectors.toSet());
        if (ObjectUtils.isEmpty(videoIdSet)) {
            log.error("随机标签库下没有数据");
        }
        return videoIdSet;
    }

    @Override
    public void pushSystemTypeStockIn(Video video) { // 将视频加入系统分类视频库
        Long typeId = video.getTypeId();
        Long videoId = video.getId();
        redisTemplate.opsForSet().add(RedisConstant.SYSTEM_TYPE_STOCK + typeId, videoId);
    }

    @Override
    public void pushSystemStockIn(Video video) { // 将视频加入系统视频库
        List<String> buildLabel = video.buildLabel();
        Long videoId = video.getId();
        redisTemplate.executePipelined((RedisCallback<?>) connection -> {
            for (String label : buildLabel) {
                String key = RedisConstant.SYSTEM_STOCK + label;
                connection.sAdd(key.getBytes(), videoId.toString().getBytes());
            }
            return null;
        });
    }

    @Override
    public void deleteSystemTypeStockIn(Video video) { // 删除系统分类库中的指定视频
        Long typeId = video.getTypeId();
        Long videoId = video.getId();
        redisTemplate.opsForSet().remove(RedisConstant.SYSTEM_TYPE_STOCK + typeId, videoId);
    }

    @Override
    public void deleteSystemStockIn(Video video) { // 删除系统视频库中的指定视频
        List<String> buildLabel = video.buildLabel();
        Long videoId = video.getId();
        redisTemplate.executePipelined((RedisCallback<?>) connection -> {
            for (String label : buildLabel) {
                String key = RedisConstant.SYSTEM_STOCK + label;
                connection.sRem(key.getBytes(), videoId.toString().getBytes());
            }
            return null;
        });
    }
}
